package com.yangye.niodemo.rpc.transport;

import com.yangye.niodemo.rpc.protocol.RpcBody;
import com.yangye.niodemo.rpc.protocol.RpcHeader;
import com.yangye.niodemo.rpc.util.SerDerUtil;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.PooledByteBufAllocator;
import io.netty.channel.socket.nio.NioSocketChannel;

import java.net.InetSocketAddress;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;

/**
 * @author yangye
 * @date 2020/12/29
 */
public class ClientFactory {

	public static final ClientFactory facotry;

	static {
		facotry = new ClientFactory();
	}

	public static final Map<InetSocketAddress, ClientPool> outboxes = new ConcurrentHashMap<>();

	public static CompletableFuture transport(RpcBody body) {
		CompletableFuture res = new CompletableFuture();
		byte[] bodyBytes = SerDerUtil.ser(body);
		RpcHeader header = RpcHeader.createHeader(bodyBytes.length);
		byte[] headerBytes = SerDerUtil.ser(header);

		ClientPool clientPool = facotry.getClientPool(new InetSocketAddress("localhost", 9090));
		NioSocketChannel client = clientPool.chooseClient();
		ByteBuf byteBuf = PooledByteBufAllocator.DEFAULT.directBuffer(headerBytes.length + bodyBytes.length);


		ResponseCallback.add(header.getRequestId(), res);
		byteBuf.writeBytes(headerBytes);
		byteBuf.writeBytes(bodyBytes);

		client.writeAndFlush(byteBuf);
		return res;
	}

	public ClientPool getClientPool(InetSocketAddress address) {
		ClientPool clientPool = outboxes.get(address);
		if (clientPool != null) {
			return clientPool;
		}
		synchronized (outboxes) {
			clientPool = outboxes.get(address);
			if (clientPool == null) {
				clientPool = new ClientPool();
				outboxes.putIfAbsent(address, clientPool);
			}
			return clientPool;
		}
	}
}
